package source;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.junit.Test;

public class Flink_Source_UserDefined {
    //TODO 自定义数据
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env.addSource(new SourceFunction<String>() {
            private Boolean isTrue=true;
            private Integer i = 0;
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while(isTrue){
                    ctx.collect("create"+ i++);
                    Thread.sleep(2000);
                }
            }

            @Override
            public void cancel() {
                this.isTrue=false;
            }
        }).print();
        env.execute();
    }
}
